package com.chencong.online.function;

import com.chencong.online.bean.AdClickCountBean;
import com.chencong.online.utils.TimestampTransformUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @program: user-behavior-analysis-online
 * @ClassName AdClinkWindowFunc
 * @description:窗口函数，将数据按照格式输出
 * @author: chencong
 * @create: 2021-12-29 15:22
 **/
public class AdClinkWindowFunc implements WindowFunction<Long, AdClickCountBean,String, TimeWindow> {
    @Override
    public void apply(String province, TimeWindow window, Iterable<Long> input, Collector<AdClickCountBean> out) throws Exception {
        long End = window.getEnd();
        String windowEnd = TimestampTransformUtil.milliTimestampToLocalDateTime(End).toString();
        Long count = input.iterator().next();
        out.collect(new AdClickCountBean(province,windowEnd,count));
    }
}
